using System;
using System.Collections.Generic;
using System.Data.Services.Client;
using System.Linq;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reflection;
using System.Reflection.Emit;
using System.Threading;
using RxHeat.Common;
using RxHeat.Tracing;

namespace RxHeat.OData
{
    /// <summary>
    ///     An Observable object that would request and pull paged results from an OData query.
    ///     The results are then pushed to observers for further processing.
    ///     Recommended for optimizing performance of asynchronous handling of large result sets over small pages.
    /// </summary>
    /// <typeparam name="TEntry">The type of the entry.</typeparam>
    public sealed class ODataObservableResults<TEntry> : IObservable<TEntry>
    {
        private readonly dynamic _context;
        private readonly IQueryable<TEntry> _query;
        private readonly IRxHeatTraceListner _rxHeatTraceListner;
        private readonly IScheduler _threadODataSubscription;
        private readonly string _tracerKey;
        private Assembly _callingAssembly;

        /// <summary>
        ///     Initializes a new instance of the <see cref="ODataObservableResults{TEntry}" /> class.
        /// </summary>
        /// <param name="query">The query.</param>
        /// <param name="context">The context.</param>
        public ODataObservableResults(IQueryable<TEntry> query, DataServiceContext context)
            : this(query,
                   context,
                   new EventLoopScheduler(st => new Thread(st)
                       {
                           Name = "ODataObservableResults Results Scheduler HC:" + context.GetHashCode(),
                           IsBackground = true,
                           Priority = ThreadPriority.Lowest
                       }))
        {
            _rxHeatTraceListner = TraceManager.GetTracer(_tracerKey);
        }

        /// <summary>
        ///     Initializes a new instance of the <see cref="ODataObservableResults{TEntry}" /> class.
        /// </summary>
        /// <param name="query">The query.</param>
        /// <param name="context">The context.</param>
        /// <param name="threadODataSubscription">
        ///     The <see cref="Scheduler" /> used for the OData query.
        /// </param>
        public ODataObservableResults(IQueryable<TEntry> query, DataServiceContext context,
                                      IScheduler threadODataSubscription)
        {
            _tracerKey = GetType().DescribeType();
            _rxHeatTraceListner = TraceManager.GetTracer(_tracerKey);
            _query = query;
            _context = context;
            _callingAssembly = Assembly.GetCallingAssembly();
            _threadODataSubscription = threadODataSubscription;
        }



        /// <summary>
        ///     Subscribes the specified observer.
        /// </summary>
        /// <param name="observer">The observer.</param>
        /// <returns></returns>
        public IDisposable Subscribe(IObserver<TEntry> observer)
        {
            IEnumerable<TEntry> response;
            try
            {
                dynamic dataServiceQuery = _query;
                response = dataServiceQuery.Execute();
                if (response == null)
                {
                    return Disposable.Empty;
                }
            }
            catch (Exception ex)
            {
                _rxHeatTraceListner.Error(ex,
                                          new TraceDetails(
                                              _tracerKey,
                                              Thread.CurrentThread.ToTraceString(_tracerKey),
                                              () => String.Format("Error during executing {0}", _query)));
                return Disposable.Empty;
            }

            var initialState = new State
                {
                    CanContinue = true,
                    Response = response
                };
            IDisposable subscriptionDisposbable = null;
            try
            {
                IObservable<TEntry> sequence = Observable.Generate(
                    initialState,
                    state => state.CanContinue,
                    MoveToNextState,
                    GetCurrentValue,
                    _threadODataSubscription).Merge();
                subscriptionDisposbable = sequence.Subscribe(_tracerKey, observer);
                string contextInstanceHC = _context.GetHashCode().ToString();
                subscriptionDisposbable.TraceDisposable(_tracerKey, "DataServiceContext Context HC:" + contextInstanceHC);
            }
            catch (Exception ex)
            {
                observer.OnError(ex);
            }
            return new CompositeDisposable(initialState, subscriptionDisposbable);

        }

        private static IObservable<TEntry> GetCurrentValue(State state)
        {
            if (state.Response == null)
                return Observable.Empty<TEntry>();
            else return state.Response.ToObservable();
        }

        private State MoveToNextState(State state)
        {
            dynamic responseType = state.Response;
            dynamic continuation = responseType.GetContinuation();
            if (continuation == null)
            {
                state.CanContinue = false;
                return state;
            }

            dynamic response;
            try
            {
                response = _context.Execute(continuation);
            }
            catch (Exception ex)
            {
                state.CanContinue = false;
                return state;
            }
            state.Response = response;
            return state;
        }

        private sealed class State : IDisposable
        {
            public bool CanContinue { get; set; }

            public IEnumerable<TEntry> Response { get; set; }

            public void Dispose()
            {
                CanContinue = false;
            }
        }
    }
}